package com.codejiwei.flink.practice;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashSet;


/**
 * @author jiwei
 * @description 统计网站独立访客数(PV)
 * @date 2023/5/23 8:51
 */
public class Flink_Project_Practice_UV {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        env.readTextFile("E:\\workspace\\ss-flink\\src\\main\\resources\\data\\UserBehavior.csv")
                .flatMap((String line, Collector< Tuple2<String, Long> > out) -> {
                    String[] fields = line.split(",");
                    String behavior = fields[3];
                    Long userId = Long.valueOf(fields[0]);
                    out.collect(Tuple2.of(behavior, userId));
                }).returns(Types.TUPLE(Types.STRING, Types.LONG))
                .filter(t -> "pv".equals(t.f0))
                .keyBy(t -> t.f0)
                .process(new KeyedProcessFunction<String, Tuple2<String, Long>, Integer>() {
                    HashSet<Long> ids = new HashSet<>();
                    @Override
                    public void processElement(Tuple2<String, Long> stringLongTuple2, Context context, Collector<Integer> collector) throws Exception {
                        ids.add(stringLongTuple2.f1);
                        collector.collect(ids.size());
                    }
                })
                .print();

        env.execute();
    }
}
